package com.xhwl.centiotdata.mq.comsumer;

import com.alibaba.fastjson.JSON;
import com.xhwl.centiotdata.handler.org.OrgFactory;
import com.xhwl.centiotdata.handler.org.OrgHandler;
import com.xhwl.centiotdata.pojo.dto.pedestrian.PedestrianIntegrationConfigDTO;
import com.xhwl.common.pojo.cent.sdata.Organization;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;

/**
 * Falcon.cao @Copyright (C), 2022, 兴海物联
 * Date: 2022/7/4 13:57
 * Description: 包信息
 */
@Component
@Slf4j
public class OrganizationConsumer {
    public static final String ORGANIZATION_CREATE = "data-core-organization-create";
    public static final String ORGANIZATION_UPDATE = "data-core-organization-update";
    public static final String ORGANIZATION_DELETE = "data-core-organization-delete";
    @KafkaListener(
            topics = {
                    ORGANIZATION_CREATE,
                    ORGANIZATION_UPDATE,
                    ORGANIZATION_DELETE
            },groupId = "IOT-DATA-ORG"
    )
    public void onReceive(ConsumerRecord<String, String> record) {
        log.info("topic:---------" + record.topic() + "----------------------------------------------------\n" + record.value());
        String content = record.value();
        List<Organization> organizations = JSON.parseArray(content, Organization.class);
        for (Organization organization : organizations) {
            List<Map<PedestrianIntegrationConfigDTO, OrgHandler>> configList = OrgFactory.getInvokeStrategyByEnterpriseId(organization.getEnterpriseId());
            if (!CollectionUtils.isEmpty(configList)) {
                configList.forEach(config -> {
                    for (Map.Entry<PedestrianIntegrationConfigDTO, OrgHandler> entry : config.entrySet()) {
                        switch (record.topic()) {
                            case ORGANIZATION_CREATE: {
                                entry.getValue().createOrg(entry.getKey(), organizations);
                                break;
                            }
                            case ORGANIZATION_UPDATE: {
                                entry.getValue().updateOrg(entry.getKey(), organizations);
                                break;
                            }
                            case ORGANIZATION_DELETE: {
                                entry.getValue().deleteOrg(entry.getKey(), organizations);
                                break;
                            }
                            default:
                                break;
                        }
                    }
                });
            }
        }
    }
}
